package main

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"mag/internal/database"
	"time"

	"github.com/emirpasic/gods/sets/hashset"
	_ "github.com/go-sql-driver/mysql"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

func main() {
	log.Println("开始导入数据")
	originInDict := make(map[string]*hashset.Set)
	originOutDict := make(map[string]*hashset.Set)

	const (
		USERNAME = "sunny"
		PASSWORD = "sunny2020"
		NETWORK  = "tcp"
		SERVER   = "192.168.1.220"
		PORT     = 23306
		DATABASE = "patents"
	)
	dsn := fmt.Sprintf("%s:%s@%s(%s:%d)/%s", USERNAME, PASSWORD, NETWORK, SERVER, PORT, DATABASE)
	DB, err := sql.Open("mysql", dsn)
	if err != nil {
		fmt.Printf("Open mysql failed,err:%v\n", err)
		return
	}
	DB.SetConnMaxLifetime(100 * time.Second) //最大连接周期，超过时间的连接就close
	DB.SetMaxOpenConns(100)                  //设置最大连接数
	DB.SetMaxIdleConns(16)                   //设置闲置连接数

	if err := DB.Ping(); err != nil {
		fmt.Println("open database fail")
		return
	}
	fmt.Println("connnect success")

	row, err := DB.Query("select PatPublNr,CitedPatPublNr from Patents_BackwardCitations")
	if err != nil {
		fmt.Println(err)
	}
	// var PatPublNr string
	for row.Next() {
		var pn, cn string
		row.Scan(&pn, &cn)

		if originInDict[cn] == nil {
			originInDict[cn] = hashset.New()
		}
		originInDict[cn].Add(pn)

		if originOutDict[pn] == nil {
			originOutDict[pn] = hashset.New()
		}
		originOutDict[pn].Add(cn)
	}

	log.Println("导入完成，开始计算")
	// 开始计算

	outchan := calDisruption(originOutDict, originInDict)
	handleOutChan(outchan, "disruption")
	log.Println("over")

}

type disruptionItem struct {
	ID string  `bson:"_id"`
	D  float64 `bson:"d"`
	I  int64   `bson:"i"`
	J  int64   `bson:"j"`
	K  int64   `bson:"k"`
}

func (c *disruptionItem) CalD() {
	c.D = (float64(c.I) - float64(c.J)) / (float64(c.I) + float64(c.J) + float64(c.K))
}

func calDisruption(outMap, inMap map[string]*hashset.Set) chan *disruptionItem {
	outChan := make(chan *disruptionItem, 10000)
	go func() {
		for key, outSet := range outMap {
			retData := disruptionItem{ID: key}

			// 获取 linksin set, 如果没有了，就跳过
			inSet := inMap[key]
			if inSet == nil || outSet.Size() == 0 {
				continue
			}

			outInSet := hashset.New()

			for _, item := range outSet.Values() {
				if outinset, ok := inMap[item.(string)]; ok {
					outInSet.Add(outinset.Values()...)
				}
			}

			// 计算颠覆度参数
			for _, nodeID := range inSet.Values() {
				nodeIDOutSet := outMap[nodeID.(string)]
				if Is_intersection(nodeIDOutSet, outSet) {
					retData.J += 1
				} else {
					retData.I += 1
				}
			}
			// 排除自身节点
			outInSet.Remove(key)
			retData.K = Cal_uniq_count(outInSet, inSet)

			retData.CalD()
			outChan <- &retData
		}
		close(outChan)
	}()
	// 没有 linksOut的，统一为 1
	go func() {
		for key, inSet := range inMap {
			if _, ok := outMap[key]; !ok {
				outChan <- &disruptionItem{
					ID: key,
					I:  int64(inSet.Size()),
					D:  1,
				}
			}
		}
	}()
	return outChan
}

func handleOutChan(outchan chan *disruptionItem, collectionName string) {

	collection := database.InitMongo().Database("prtent").Collection(collectionName)

	opts := options.BulkWrite().SetOrdered(false)
	models := []mongo.WriteModel{}
	i := 0
	for item := range outchan {
		i += 1
		models = append(models, mongo.NewInsertOneModel().SetDocument(item))
		if i%10000 == 0 {
			retTmp, err := collection.BulkWrite(context.Background(), models, opts)
			if err != nil {
				log.Println("bulk upsert fail", err)
				log.Printf("%+v", retTmp)
			}
			models = []mongo.WriteModel{}
		}
	}
	if len(models) > 0 {
		retTmp, err := collection.BulkWrite(context.Background(), models, opts)
		if err != nil {
			log.Println("bulk upsert fail", err)
			log.Printf("%+v", retTmp)
		}
	}
}

// 计算是否有交集
func Is_intersection(a, b *hashset.Set) bool {
	for _, v := range a.Values() {
		if b.Contains(v.(string)) {
			return true
		}
	}
	return false
}

// 计算差
func Cal_uniq_count(a, b *hashset.Set) (count int64) {
	for _, v := range a.Values() {
		if !b.Contains(v.(string)) {
			count += 1
		}
	}
	return
}
